package org.lzfto.flink.demo.business.flinkTest.flinkObjects.function.map;

import org.apache.flink.api.common.functions.MapFunction;
import org.lzfto.flink.demo.business.orders.OrderCommon;
import org.lzfto.flink.demo.business.orders.OrderKey;
import org.lzfto.flink.demo.business.orders.SimpleOrder;
import org.lzfto.flink.demo.business.orders.SimpleOrderAgg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson2.JSON;

/**
*把加盐的汇总数据转为不加盐的数据. 要求输出的和输出的保持一直
*@author lzfto
*@创建时间 2025年6月24日 11:47:41
*/
public class OrderDeSaltMapFun<T extends OrderKey> implements MapFunction<T, T> {
	private static final long serialVersionUID = 1L;
	private final  Logger LOGGER = LoggerFactory.getLogger(getClass());
	
	private String grpType;
	
	public  OrderDeSaltMapFun(String grpType) {
		this.grpType=grpType;
		LOGGER.info("创建去盐函数,分组类型:{}",grpType);
	}
	
	@SuppressWarnings("unchecked")
	@Override
	public T map(T value) throws Exception {
		if (value instanceof SimpleOrder order ) {
			String before=JSON.toJSONString(order);
			if (grpType.equals(OrderCommon.GRP_TYPE_SEX)) {
				String sex = order.getSex().substring(0, order.getSex().lastIndexOf("-"));
				order.setSex(sex);

			} else if (grpType.equals(OrderCommon.GRP_TYPE_ZONE)) {
				String zoneName = order.getZoneName().substring(0, order.getZoneName().lastIndexOf("-"));
				order.setZoneName(zoneName);
			} else {
				String sex = order.getSex().substring(0, order.getSex().lastIndexOf("-"));
				order.setSex(sex);
			}
			LOGGER.info("去盐前后:{},{}",before,JSON.toJSONString(order));
			return (T) order;
		}
		else if (value instanceof SimpleOrderAgg agg  ) {
			String before=JSON.toJSONString(agg);
			if (grpType.equals(OrderCommon.GRP_TYPE_SEX)) {
				String sex = agg.getSex().substring(0, agg.getSex().lastIndexOf("-"));
				agg.setSex(sex);

			} else if (grpType.equals(OrderCommon.GRP_TYPE_ZONE)) {
				String zoneName = agg.getZoneName().substring(0, agg.getZoneName().lastIndexOf("-"));
				agg.setZoneName(zoneName);
			} else {
				String sex = agg.getSex().substring(0, agg.getSex().lastIndexOf("-"));
				agg.setSex(sex);
			}
			LOGGER.info("去盐前后:{},{}",before,JSON.toJSONString(agg));
			return (T) agg;
		}
		else {
			return null;
		}
		
	}

}
